package com.storm.lesson.transaction1;

import java.math.BigInteger;

import backtype.storm.transactional.ITransactionalSpout.Coordinator;
import backtype.storm.utils.Utils;

/**
 * 
 * @author zhihu
 *
 */
public class MyCoordinator implements Coordinator<MyMata> {

	public static int BATCH_NUM = 10;

	/**
	 * 创建一个新的metadata，当isReady() 为true时，发射该metadata（事务tuple）到“batch emit”流
	 */
	@Override
	public MyMata initializeTransaction(BigInteger txid, MyMata prevMetadata) {
		long beginPoint = 0;
		if (prevMetadata == null) {
			beginPoint = 0;
		} else {
			beginPoint = prevMetadata.getBeginPoint() + prevMetadata.getNum();
		}

		MyMata mata = new MyMata();
		mata.setBeginPoint(beginPoint);
		mata.setNum(BATCH_NUM);
		System.err.println("启动一个事务: " + mata.toString());

		return mata;
	}

	/**
	 * 为true时启动新事务，需要时可以在此sleep
	 */
	@Override
	public boolean isReady() {
		Utils.sleep(2000);
		return true;
	}

	@Override
	public void close() {
		// TODO Auto-generated method stub

	}

}
